package com.killer.auth.util;

import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;

import java.util.function.BiConsumer;

@Slf4j
public class BulkUtil {

    public static BulkProcessor getBulkProcessor(RestHighLevelClient restHighLevelClient) {
        BulkProcessor bulkProcessor=null;
        BulkProcessor.Listener listener=new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                log.info("Try to insert data number:"+request.numberOfActions());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                log.info("success insert data number:"+request.numberOfActions()+",id:"+executionId);
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                log.error("Bulk is unsuccess:"+failure+",id:"+executionId);
            }
        };
        BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer=new BiConsumer<BulkRequest, ActionListener<BulkResponse>>() {
            @Override
            public void accept(BulkRequest bulkRequest, ActionListener<BulkResponse> bulkResponseActionListener) {
                restHighLevelClient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener);

            }
        };
        BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
        builder.setBulkActions(5000);
        builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
        builder.setConcurrentRequests(10);
        builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
        bulkProcessor=builder.build();
        return bulkProcessor;
    }
}
